package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo2CLazzNum {
  def main(args: Array[String]): Unit = {
    //1、创建spark配置文件对象
    val conf = new SparkConf()
    //设置spark运行环境  local[2]  模拟两个线程
    conf.setMaster("local[2]")
    //设置job名
    conf.setAppName("Demo2CLazzNum")

    //2、创建spark上下文对象  spark编程的入口
    val sc = new SparkContext(conf)

    //3、读取数据  构建第一个和RDD
    val studentsRDD: RDD[String] = sc.textFile("spark/data/students.txt")

    /**
      * 统计每个班级人数
      *
      */

    //1、取出班级
    val clazzRDD: RDD[String] = studentsRDD.map(line => {
      val clazz = line.split(",")(4)
      clazz
    })

    //2、通过班级进行分组
    val groupRDD: RDD[(String, Iterable[String])] = clazzRDD.groupBy(clazz => clazz)

    //3、统计人数
    val sumRDD: RDD[String] = groupRDD.map(kv => {
      val clazz = kv._1

      var sum = 0
      for (i <- kv._2) {
        sum = sum + 1
      }

      clazz + "\t" + sum
    })

    //spark程序最后需要一个action算子
    sumRDD.foreach(println)

  }
}
